package io.plus.interactive;

import drds.common.Clock;
import io.plus.interactive.task.CloseCode;
import io.plus.interactive.utils.ByteBufferUtils;
import io.plus.interactive.utils.IoUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.nio.ByteBuffer;
import java.nio.channels.CompletionHandler;

@Data
@Slf4j
public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {

    private AsynchronousSocketChannelContext asynchronousSocketChannelContext = null;
    private ByteBuffer readByteBuffer;

    public ReadCompletionHandler(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        this.asynchronousSocketChannelContext = asynchronousSocketChannelContext;
        this.readByteBuffer = ByteBuffer.allocate(asynchronousSocketChannelContext.getReadBufferSize());
        this.readByteBuffer.order(asynchronousSocketChannelContext.clientConfiguration.getByteOrder());
    }

    @Override
    public void completed(Integer result, ByteBuffer byteBuffer) {
        if (result == 0) {
            log.error("{}, 读到的数据长度为0", asynchronousSocketChannelContext);
            Io.close(asynchronousSocketChannelContext, null, "读到的数据长度为0", CloseCode.READ_COUNT_IS_ZERO);
            return;
        } else if (result < 0) {
            if (result == -1) {
                Io.close(asynchronousSocketChannelContext, null, "对方关闭了连接", CloseCode.CLOSED_BY_PEER);
                return;
            } else {
                Io.close(asynchronousSocketChannelContext, null, "读数据时返回" + result, CloseCode.READ_COUNT_IS_NEGATIVE);
                return;
            }
        } else if (result > 0) {
            asynchronousSocketChannelContext.asynchronousSocketChannelState.latestTimeOfReceivedByte = Clock.currentTimeMillis;
            readByteBuffer.flip();
            {
                if (asynchronousSocketChannelContext.clientConfiguration.useQueueDecode) {
                    asynchronousSocketChannelContext.decodeTask.add(ByteBufferUtils.copy(readByteBuffer));
                    asynchronousSocketChannelContext.decodeTask.execute();
                } else {
                    asynchronousSocketChannelContext.decodeTask.setNewReceivedByteBuffer(readByteBuffer);
                    asynchronousSocketChannelContext.decodeTask.decode();
                }
            }
            if (IoUtils.check(asynchronousSocketChannelContext)) {
                read();
            }

        }
    }

    private void read() {
        if (readByteBuffer.capacity() == asynchronousSocketChannelContext.getReadBufferSize()) {
            readByteBuffer.position(0);
            readByteBuffer.limit(readByteBuffer.capacity());
        } else {
            readByteBuffer = ByteBuffer.allocate(asynchronousSocketChannelContext.getReadBufferSize());
        }

        asynchronousSocketChannelContext.asynchronousSocketChannel.read(readByteBuffer, readByteBuffer, this);
    }

    @Override
    public void failed(Throwable throwable, ByteBuffer byteBuffer) {
        Io.close(asynchronousSocketChannelContext, throwable, "读数据时发生异常: " + throwable.getClass().getName(), CloseCode.READ_ERROR);
    }

}
